package com.traffic.visim.socket;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.traffic.visim.common.CommonConstant;
import com.traffic.visim.kafkamq.constant.KafkaConstant;
import com.traffic.visim.kafkamq.register.KafkaProducer;
import com.traffic.visim.onvif.dto.SignInfo;
import com.traffic.visim.onvif.mapper.IOnvifEventMapper;
import com.traffic.visim.service.OnvifService;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;


import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Arrays;
import java.util.List;
import java.util.Map;


@Slf4j
@AllArgsConstructor
public class TestSocketService implements Runnable{

    private Socket socket;

    private KafkaProducer kafkaProducer;



    private IOnvifEventMapper onvifEventMapper;

    private StringRedisTemplate stringRedisTemplate;

    OnvifService onvifService;

    @Override
    @SneakyThrows
    public void run() {
        InputStream inputStream = socket.getInputStream();
        OutputStream outputStream = socket.getOutputStream();
        byte[] buffer = new byte[1024];
        int len;
        log.info("handleClient 方法");
        while ((len = inputStream.read(buffer)) != -1) {
            String hexString = bytesToHexString(buffer, len);
            log.info("hexString is {}",hexString);
            String ip=socket.getInetAddress().getHostAddress();
            Integer type=0;
            if (hexString.contains("31"))
            {
                type=1;
            }
            else if(hexString.contains("32"))
            {
                type=2;
            }
            //闯红灯
            else if (hexString.contains("33"))
            {
                type=4;
            }
            //违章停车
            else if (hexString.contains("34"))
            {
                type=3;
            }
            if (3==type)
            {
                List<SignInfo> infoList= onvifEventMapper.getSignInfoByMonitorIP(ip);
                if (CollUtil.isNotEmpty(infoList))
                {
                    SignInfo info=infoList.get(0);
                    Integer number= info.getNumber();
                    String monitorInfo=info.getMonitorInfo();
                    String direction="";
                    if (ObjectUtil.isNotEmpty(monitorInfo)&&monitorInfo.indexOf("_")!=0)
                    {
                        String[] str=monitorInfo.split("_");
                        direction=str[str.length-1];
                    }
                    Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(CommonConstant.REDIS_CROSS_KEY + number);
                    if (ObjectUtil.isNotEmpty(map))
                    {
                        String curstg = (String) map.get(CommonConstant.CURSTG);
                        //南北直行绿灯
                        if ("1".equals(curstg)&&ObjectUtil.isNotEmpty(direction)&&(direction.equals("西")||direction.equals("东")))
                        {
                            Map map1= Maps.newHashMap();
                            map1.put("ip",socket.getInetAddress().getHostAddress());
                            map1.put("type",type);
//                            kafkaProducer.send(KafkaConstant.KAFKA_TRAFFIC_TOPIC, "event",JSON.toJSONString(map1));
                            onvifService.sendEvent(socket.getInetAddress().getHostAddress(),type);
                        }
                        //东西直行
                        else if ("2".equals(curstg)&&ObjectUtil.isNotEmpty(direction)&&(direction.equals("南")||direction.equals("北")))
                        {
                            Map map1= Maps.newHashMap();
                            map1.put("ip",socket.getInetAddress().getHostAddress());
                            map1.put("type",type);
//                            kafkaProducer.send(KafkaConstant.KAFKA_TRAFFIC_TOPIC, "event",JSON.toJSONString(map1));
                            onvifService.sendEvent(socket.getInetAddress().getHostAddress(),type);
                        }
                    }
                }


            }
            else {
                Map map= Maps.newHashMap();
                map.put("ip",socket.getInetAddress().getHostAddress());
                map.put("type",type);
//                kafkaProducer.send(KafkaConstant.KAFKA_TRAFFIC_TOPIC, "event",JSON.toJSONString(map));
                onvifService.sendEvent(socket.getInetAddress().getHostAddress(),type);
            }




        }
        inputStream.close();
        outputStream.close();
        socket.close();
    }

    public static String bytesToHexString(byte[] bytes, int len) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < len; i++) {
            sb.append(String.format("%02X ", bytes[i]));
        }
        return sb.toString();
    }
}
